package org.apache.ocean.main.replication;

import java.io.File;
import java.io.IOException;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.NavigableSet;
import java.util.Set;
import java.util.SortedMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.ocean.main.fields.NumberUtils;
import org.apache.ocean.main.replication.RawLogFile.Data;
import org.apache.ocean.main.replication.RawLogFile.Load;
import org.apache.ocean.main.replication.RawLogFile.RecordHeader;
import org.apache.ocean.main.util.OceanSkipList;

public class LogFile2 {
	private ConcurrentHashMap<String, OceanSkipList<Long, RecordHeader>> categorySnapshotMap = new ConcurrentHashMap<String, OceanSkipList<Long, RecordHeader>>();
	private OceanSkipList<Long, RecordHeader> snapshotMap = new OceanSkipList<Long, RecordHeader>();
	private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
	private RawLogFile rawLogFile;
	private Long id;

	public LogFile2(Long id, File file) throws IOException {
		this.id = id;
		rawLogFile = new RawLogFile(file);
		List<RecordHeader> recordHeaders = rawLogFile.loadRecordHeaders();
		for (RecordHeader recordHeader : recordHeaders) {
			addRecordHeader(recordHeader);
		}
	}
  
	public boolean checksum(String category, Long lowId, Long highId, MessageDigest messageDigest) {
		OceanSkipList<Long, RecordHeader> sortedMap = categorySnapshotMap.get(category);
		NavigableSet<Long> set = sortedMap.subMap(lowId, true, highId, true).navigableKeySet();
		return checksum(messageDigest, set);
	}
	
	public boolean checksum(Long lowId, Long highId, MessageDigest messageDigest) {
		NavigableSet<Long> set = snapshotMap.subMap(lowId, true, highId, true).navigableKeySet();
		return checksum(messageDigest, set);
	}
	
	public boolean checksum(MessageDigest messageDigest, Set<Long> set) {
		byte[] buffer = new byte[8];
		for (Long key : set) {
			NumberUtils.encode(key, buffer);
			messageDigest.digest(buffer);
		}
		return true;
	}
	
	public long size() {
		return rawLogFile.size();
	}
	
	public boolean delete(Long id) throws IOException {
		lock.writeLock().lock();
		try {
			RecordHeader recordHeader = snapshotMap.get(id);
			boolean deleted = rawLogFile.delete(recordHeader);
			if (deleted) {
				snapshotMap.remove(id);
				for (ConcurrentSkipListMap<Long, RecordHeader> categoryMap : categorySnapshotMap.values()) {
					if (categoryMap.containsKey(id)) {
						categoryMap.remove(id);
					}
				}
			}
			return deleted;
		} finally {
			lock.writeLock().unlock();
		}
	}

	public Long getId() {
		return id;
	}

	public Long getPreviousId(Long id) {
		SortedMap<Long, RecordHeader> headMap = snapshotMap.headMap(id);
		return headMap.lastKey();
	}

	public Long getLowestId() {
		return snapshotMap.firstKey();
	}

	public boolean containsId(Long id) {
		return snapshotMap.containsKey(id);
	}

	public Long getHighestId() {
		List<Long> list = new LinkedList<Long>();
		for (OceanSkipList<Long, RecordHeader> map : categorySnapshotMap.values()) {
			Long lastKey = map.lastKey();
			if (lastKey != null)
				list.add(lastKey);
		}
		return Collections.max(list);
	}

	public boolean containsId(String category, Long id) {
		OceanSkipList<Long, RecordHeader> categoryMap = getCategoryMap(category);
		if (categoryMap == null)
			return false;
		return categoryMap.containsKey(id);
	}

	public List<RecordHeader> getRecordHeaders(Long fromKey, int num) {
		ConcurrentNavigableMap<Long,RecordHeader> subMap = snapshotMap.tailMap(fromKey);
		Iterator<RecordHeader> iterator = subMap.values().iterator();
		int count = 0;
		List<RecordHeader> recordHeaders = new ArrayList<RecordHeader>(num);
		while (count < num && iterator.hasNext()) {
			RecordHeader recordHeader = iterator.next();
			recordHeaders.add(recordHeader);
			count++;
		}
		return recordHeaders;
	}
  
	public void commit(RecordHeader recordHeader) {
		addRecordHeader(recordHeader);
	}
	
	public RecordHeader writeRecord(Long id, String category, long schemaVersion, Data data) throws IOException {
		lock.writeLock().lock();
		try {
			RecordHeader recordHeader = rawLogFile.write(id, category, schemaVersion, data);
			addRecordHeader(recordHeader);
			return recordHeader;
		} finally {
			lock.writeLock().unlock();
		}
	}

	private void addRecordHeader(RecordHeader recordHeader) {
		snapshotMap.put(recordHeader.id, recordHeader);
		OceanSkipList<Long, RecordHeader> categoryMap = getCategoryMap(recordHeader.category);
		categoryMap.put(recordHeader.id, recordHeader);
	}

	public Record getRecordGreaterOrEqualToID(EnumSet<Load> load, Long id, ByteBufferPool documentByteBuffers, ByteBufferPool deletesByteBuffer) throws IOException {
		RecordHeader recordHeader = snapshotMap.getGreaterOrEqualTo(id);
		if (recordHeader == null)
			return null;
		return getRecord(load, recordHeader, documentByteBuffers, deletesByteBuffer);
	}

	public Record getRecord(EnumSet<Load> load, RecordHeader recordHeader, ByteBufferPool documentByteBuffers, ByteBufferPool deletesByteBuffer) throws IOException {
		Data data = rawLogFile.readData(load, recordHeader, documentByteBuffers, deletesByteBuffer);
		Record record = new Record(recordHeader.id, recordHeader.category, recordHeader.schemaVersion, data);
		return record;
	}

	public Record getRecordGreaterOrEqualToID(EnumSet<Load> load, String category, Long id, ByteBufferPool documentByteBuffers, ByteBufferPool deletesByteBuffer) throws IOException {
		OceanSkipList<Long, RecordHeader> map = categorySnapshotMap.get(category);
		RecordHeader recordHeader = map.getGreaterOrEqualTo(id);
		if (recordHeader == null)
			return null;
		return getRecord(load, recordHeader, documentByteBuffers, deletesByteBuffer);
	}

	public Long getLowestID(String category) {
		OceanSkipList<Long, RecordHeader> categoryMap = categorySnapshotMap.get(category);
		if (categoryMap == null)
			throw new IllegalArgumentException("category: " + category + " unknown");
		return categoryMap.firstKey();
	}

	public Long getHighestID(String category) {
		OceanSkipList<Long, RecordHeader> categoryMap = categorySnapshotMap.get(category);
		if (categoryMap == null)
			throw new IllegalArgumentException("category: " + category + " unknown");
		return categoryMap.lastKey();
	}

	public OceanSkipList<Long, RecordHeader> getCategoryMap(String category) {
		OceanSkipList<Long, RecordHeader> snapshotMap = categorySnapshotMap.get(category);
		if (snapshotMap == null) {
			snapshotMap = new OceanSkipList<Long, RecordHeader>();
			categorySnapshotMap.put(category, snapshotMap);
		}
		return snapshotMap;
	}

	public static class Record {
		private Long id;
		private String category;
		private long schemaVersion;
		private Data data;

		public Record(Long id, String category, long schemaVersion, Data data) {
			this.id = id;
			this.category = category;
			this.schemaVersion = schemaVersion;
			this.data = data;
		}

		public long getSchemaVersion() {
			return schemaVersion;
		}

		public String getCategory() {
			return category;
		}

		public Long getId() {
			return id;
		}

		public Data getData() {
			return data;
		}
	}
}
